package com.asiainfo.dacp.datastash.broker.core;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.core.enums.ErrorCodeEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.Broker;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogVo;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import com.asiainfo.dacp.datastash.broker.util.HostIpHelper;
import com.asiainfo.dacp.datastash.broker.util.ZKUtil;
import com.asiainfo.dacp.util.JsonHelper;

@Service
public class BrokerRegistryService {
	private static Logger LOG = LoggerFactory.getLogger(BrokerRegistryService.class);

	@Value(value = "${broker.id}")
	private String brokerId;

	@Autowired
	private StashTraceLogger tracelog;

	private static final String SPLIT_CHART = "/";

	private String clzName = this.getClass().getSimpleName();

	public void registry() {
		try {
			String brokerIp = HostIpHelper.getHostIp();
			Broker broker = new Broker(brokerId, brokerIp);
			String data = JsonHelper.getInstance().write(broker);

			if (ZKUtil.exists(BrokerContext.ACTIVE_BROKERS + SPLIT_CHART + brokerId) != null) {
				throw new BrokerException(ErrorCodeEnum.CONFIG_ERROR,
						String.format("ERROR: BORKER ID [%s] 已经注册，不能重复启动 . ", brokerId));
			}

			ZKUtil.createP(BrokerContext.ACTIVE_BROKERS);
			ZKUtil.createE(BrokerContext.ACTIVE_BROKERS + SPLIT_CHART + brokerId, data);

			ZKUtil.createP(BrokerContext.HB_ADDRESS);
			ZKUtil.createE(BrokerContext.HB_ADDRESS + SPLIT_CHART + brokerIp, data);

			/*
			 * CuratorFramework newClient = CuratorFrameworkFactory.newClient(
			 * BrokerContext.ZK_ADDRESS, new RetryNTimes(0, 5000));
			 * newClient.start(); this.createMultiNode(newClient,
			 * BrokerContext.HB_ADDRESS, CreateMode.PERSISTENT, "");
			 * this.createMultiNode(newClient,
			 * BrokerContext.HB_ADDRESS+SPLIT_CHART+brokerIp,
			 * CreateMode.EPHEMERAL, data);
			 */

			tracelog.info(new StashTraceLogVo(clzName, "broker:boot", "注册Broker信息成功"));
		} catch (Exception e) {
			e.printStackTrace();
			System.exit(-99);
		}
	}

	private String createMultiNode(CuratorFramework newClient, String path, CreateMode createMode, String data)
			throws Exception {
		String tempPath = "";
		if (newClient.checkExists().forPath(path) == null) {
			String[] paths = path.split("\\/");
			for (String p : paths) {
				if (!StringUtils.isEmpty(p)) {
					tempPath += "/" + p;
					if (newClient.checkExists().forPath(tempPath) == null) {
						newClient.create().withMode(createMode).forPath(tempPath, data.getBytes());
					}
				}
			}
		}
		return path;

	}

}
